package com.job.core.thread;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;

import org.eclipse.jetty.util.ConcurrentHashSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.job.core.biz.model.HandleCallBackParam;
import com.job.core.biz.model.ReturnModel;
import com.job.core.biz.model.TriggerParam;
import com.job.core.handler.BaseJobHandler;
import com.job.core.log.JobLogAppender;
import com.job.core.log.JobLogger;
/**
 * 
 * @Title: JobThread.java
 * @Package: com.job.core.thread
 * @Description: job执行线程
 * @author: sunwei
 * @date: 2017年5月10日 下午3:54:35
 * @version: V1.0
 */
public class JobThread  extends Thread{
    private Logger logger=LoggerFactory.getLogger(JobThread.class);
    private BaseJobHandler jobHandler;//要执行的执行器
    private LinkedBlockingDeque<TriggerParam> triggerQueue;//存放执行job的队列
    private ConcurrentHashSet<Long> triggerLogId;//log日志 防止重复
    private boolean toStop=false;//线程停止标志
    private String stopReason;//停止原因
    /**
     * 
     * @Description:构造函数 初始化必要参数
     * @param jobHandler 执行器
     */
    public JobThread(BaseJobHandler jobHandler) {
	this.jobHandler=jobHandler;
	triggerQueue=new LinkedBlockingDeque<TriggerParam>();
	triggerLogId=new ConcurrentHashSet<Long>();
    }
    public BaseJobHandler getHandler() {
	return jobHandler;
    }
    /**
     * 
     * @Title: pushTriggerQueue
     * @Description: 将要执行的执行器放入触发队列
     * @author:sunwei
     * @createTime:2017年5月11日上午9:20:01
     * @param triggerParam 触发参数
     */
    public void pushTriggerQueue(TriggerParam triggerParam){
	if(triggerLogId.contains(triggerParam.getLogId())){//logId作为是否执行过的标识，如果存在则表示已经执行过
	    logger.info("jobId："+triggerParam.getJobId()+",logId："+triggerParam.getLogId()+"is repeated");
	    return;
	}
	//放入队列
	triggerQueue.add(triggerParam);
	triggerLogId.add(triggerParam.getLogId());
    }
    /**
     * 
     * @Title: toStop
     * @Description:停止线程
     * @author:sunwei
     * @createTime:2017年5月11日上午9:25:03
     * @param stopReason 停止原因
     */
    public void toStop(String stopReason){
	/**
	 * Thread.interrupt只支持终止线程的阻塞状态(wait、join、sleep)，
	 * 在阻塞出抛出InterruptedException异常,改变线程状态（jdk api中说明），
	 * 但是并不会终止运行的线程本身；所以需要注意，此处彻底销毁本线程，
	 * 需要通过共享变量方式；
	 */
	this.toStop = true;
	this.stopReason = stopReason;
    }
    /**
     * 
     * @Title: run
     * @Description: job执行线程
     * @author:sunwei
     * @createTime:2017年5月11日上午11:17:13
     * @see java.lang.Thread#run()
     */
    @Override
    public void run(){
	while(!toStop){//没有停止
	    try{
		// to check toStop signal, we need cycle, so wo cannot use queue.take(), instand of poll(timeout)
		TriggerParam triggerParam=triggerQueue.poll(3L, TimeUnit.SECONDS);
		if(null!=triggerParam){//不存在
		    triggerLogId.remove(triggerParam.getLogId());//删除
		    String [] params=(null!=triggerParam.getExecutorParams() && triggerParam.getExecutorParams().trim().length()>0)?(String []) Arrays.asList(triggerParam.getExecutorParams().split(",")).toArray():null;
		    ReturnModel<String> returnModel=null;
		    //创建日志文件
		    try {
			String logFile=JobLogAppender.makeLogFileName(new Date(triggerParam.getLogDateTim()), triggerParam.getLogId());
			JobLogAppender.contextHolder.set(logFile);
			logger.info("------jobId"+triggerParam.getJobId() +"start-----");
			returnModel=jobHandler.executor(params);//执行
			if (returnModel == null) {
			returnModel = ReturnModel.FAILUE;
			}
		    } catch (Exception e) {
			if (toStop) {
				JobLogger.log("<br>----------- jobId"+triggerParam.getJobId()+" toStop, stopReason:" + stopReason);
			}
			StringWriter stringWriter = new StringWriter();
			e.printStackTrace(new PrintWriter(stringWriter));
			String errorMsg = stringWriter.toString();
			JobLogger.log("JobThread:"+triggerParam.getJobId()+" Exception:" + errorMsg);
			returnModel = new ReturnModel<String>(ReturnModel.FAILUE_CODE, stringWriter.toString());
		    }
		    JobLogger.log("------jobId"+triggerParam.getJobId() +"end-----");
		 // callback handler info
		    if (!toStop) {
			// commonm
			TriggerCallBackThread.pushCallBack(new HandleCallBackParam(triggerParam.getLogId(), triggerParam.getLogAddress(), returnModel));
		    } else {
			// is killed
			ReturnModel<String> stopResult = new ReturnModel<String>(ReturnModel.FAILUE_CODE, stopReason + " [业务运行中，被强制终止]");
			TriggerCallBackThread.pushCallBack(new HandleCallBackParam(triggerParam.getLogId(), triggerParam.getLogAddress(), stopResult));
		    }
		}
	    }catch(Exception exception){
		if (toStop) {
			JobLogger.log("<br>----------- job toStop, stopReason:" + stopReason);
		}

		StringWriter stringWriter = new StringWriter();
		exception.printStackTrace(new PrintWriter(stringWriter));
		String errorMsg = stringWriter.toString();
		JobLogger.log("----------- job JobThread Exception:" + errorMsg);
	    }
	    
	    // callback trigger request in queue
	    while (triggerQueue != null && triggerQueue.size() > 0) {
		TriggerParam triggerParam = triggerQueue.poll();
		if (triggerParam != null) {
		    // is killed
		    ReturnModel<String> stopResult = new ReturnModel<String>(ReturnModel.FAILUE_CODE, stopReason+ " [任务尚未执行，在调度队列中被终止]");
		    TriggerCallBackThread.pushCallBack(new HandleCallBackParam(triggerParam.getLogId(), triggerParam.getLogAddress(), stopResult));
		}
	    }
	    logger.info(">>>>>>>>>>>>job JobThread stoped, hashCode:{}",Thread.currentThread());
	}
    }
    
}
